package each_webcode

import com.alibaba.fastjson.JSON
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import common.HbaseTools
import scala.collection.JavaConverters.asScalaSetConverter
import scala.collection.mutable

object My_BIRD {
        def main(args: Array[String]): Unit = {
                    val conf=new SparkConf().setAppName("My_BIRD").setMaster("local[4]")
                    val sc=new SparkContext(conf)
                //    val inpath="file:///Users/kelisiduofu/file/sourcecode2.json"
                    val inpath="file:///Users/kelisiduofu/file/TEST"
                    val output="file:///Users/kelisiduofu/file/sourcecodeX1"
                    val lines=sc.textFile(inpath)

                    val pre_lines: RDD[(String, Map[String, String])] = lines.map(line => {
                      val tmp = new mutable.HashMap[String, String]()
                      val obj = JSON.parseObject(line)
                      obj.keySet().asScala.foreach(key => {
                        tmp.put(key, obj.getString(key))
                      })
                      tmp("_id") = tmp("_id").substring(9, 33)
                      (tmp("SourceCode"), tmp.toMap)
                    })

                    pre_lines.groupByKey().map{
                      case(k:String,it:mutable.Iterable[Map[String,String]])=>{
                        val tools=new HbaseTools()
                        val connection=tools.getConnection()
                        tools.getFamily_column_value(connection,"base",k,"dealTypes").toMap

                      }
                    }


                }
        }
